package ods;

import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * 模拟数据发送到 topic-log（增加用户属性字段和设备测量数据）
 */
public class dwd_user_kafka {

    private static final int NUM_RECORDS = 1000; // 生成记录数
    private static final String TOPIC_NAME = "topic-log"; // Kafka主题名称
    private static final long START_TIMESTAMP = System.currentTimeMillis(); // 起始时间戳
    private static final double OUT_OF_ORDER_RATIO = 0.2; // 乱序事件比例
    private static final int MAX_DELAY_SECONDS = 10; // 最大乱序延迟秒数
    private static final double PROFILE_UPDATE_RATIO = 0.05; // 用户属性更新概率
    private static final double BODY_SCALE_RATIO = 0.3; // 体脂秤数据比例
    private static final double VALID_DATA_RATIO = 0.9; // 有效数据比例

    // 枚举值
    private static final List<String> BEHAVIOR_TYPES = Arrays.asList(
            "browse", "search", "collect", "cart", "buy"
    );
    private static final List<String> SOURCES = Arrays.asList("APP", "PC");
    private static final List<String> PROFILE_SOURCES = Arrays.asList("会员系统", "设备");
    private static final List<String> GENDERS = Arrays.asList("男", "女", "未知");
    private static final List<String> DEVICE_TYPES = Arrays.asList("体脂秤", "手机");

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 创建自定义数据源
        DataStream<String> userBehaviorStream = env.addSource(new UserBehaviorSource());

        // 将数据流写入Kafka
        userBehaviorStream.addSink(KafkaUtil.getKafkaProducer(TOPIC_NAME));

        // 打印数据流（可选）
        userBehaviorStream.print();

        // 执行作业
        env.execute("UserBehaviorDataGenerator");
    }

    // 用户属性类
    private static class UserProfile {
        String birthday;
        String gender;
        float height;
        float weight;
        long updateTime;
        String profileSource;

        UserProfile(String birthday, String gender, float height, float weight, long updateTime, String profileSource) {
            this.birthday = birthday;
            this.gender = gender;
            this.height = height;
            this.weight = weight;
            this.updateTime = updateTime;
            this.profileSource = profileSource;
        }
    }

    // 自定义数据源函数
    public static class UserBehaviorSource implements SourceFunction<String> {
        private boolean isRunning = true;
        private final Random random = new Random();
        private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        private final Map<String, UserProfile> userProfiles = new HashMap<>();
        private final SimpleDateFormat birthdayFormat = new SimpleDateFormat("yyyy-MM-dd");

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            // 生成用户ID和商品ID的基数
            int userBase = 10000 + random.nextInt(90000);
            int itemBase = 1000 + random.nextInt(9000);
            int categoryBase = 10 + random.nextInt(90);

            long lastTimestamp = START_TIMESTAMP;
            long startBirthday = birthdayFormat.parse("1970-01-01").getTime();
            long endBirthday = birthdayFormat.parse("2010-12-31").getTime();

            for (int i = 0; i < NUM_RECORDS && isRunning; i++) {
                // 生成用户ID
                String userId = "user_" + (userBase + random.nextInt(100));

                // 获取或生成用户属性
                UserProfile profile = userProfiles.get(userId);
                if (profile == null || random.nextDouble() < PROFILE_UPDATE_RATIO) {
                    // 生成新的用户属性
                    String birthday = birthdayFormat.format(new Date(startBirthday + (long)(random.nextDouble() * (endBirthday - startBirthday))));
                    String gender = GENDERS.get(random.nextInt(GENDERS.size()));

                    float height;
                    float weight;
                    if ("男".equals(gender)) {
                        height = 160 + random.nextFloat() * 30; // 160-190cm
                        weight = 60 + random.nextFloat() * 30;  // 60-90kg
                    } else if ("女".equals(gender)) {
                        height = 150 + random.nextFloat() * 30; // 150-180cm
                        weight = 45 + random.nextFloat() * 25; // 45-70kg
                    } else {
                        height = 140 + random.nextFloat() * 50; // 140-190cm
                        weight = 40 + random.nextFloat() * 40; // 40-80kg
                    }

                    String profileSource = PROFILE_SOURCES.get(random.nextInt(PROFILE_SOURCES.size()));
                    long updateTime = System.currentTimeMillis();

                    profile = new UserProfile(birthday, gender, height, weight, updateTime, profileSource);
                    userProfiles.put(userId, profile);
                }

                // 生成行为数据
                String behavior = BEHAVIOR_TYPES.get(random.nextInt(BEHAVIOR_TYPES.size()));
                String itemId = "item_" + (itemBase + random.nextInt(500));
                String categoryId = "cate_" + (categoryBase + random.nextInt(50));
                String source = SOURCES.get(random.nextInt(SOURCES.size()));

                // 生成时间戳
                long eventTime;
                if (random.nextDouble() < OUT_OF_ORDER_RATIO) {
                    long delay = TimeUnit.SECONDS.toMillis(1 + random.nextInt(MAX_DELAY_SECONDS));
                    eventTime = lastTimestamp - delay;
                } else {
                    long increment = TimeUnit.SECONDS.toMillis(1 + random.nextInt(60));
                    eventTime = lastTimestamp + increment;
                    lastTimestamp = eventTime;
                }

                // 新增字段：设备类型、测量数据
                String deviceType = DEVICE_TYPES.get(random.nextInt(DEVICE_TYPES.size()));

                // 测量数据只在体脂秤设备上生成
                Float weightKg = null;
                Float heightCm = null;
                String measureTime = null;
                Boolean isValid = false;

                if (true) {
                    // 生成测量数据（基于用户属性值±5%波动）
                    float weightVariation = profile.weight * (0.95f + random.nextFloat() * 0.1f);
                    float heightVariation = profile.height * (0.95f + random.nextFloat() * 0.1f);

                    weightKg = weightVariation;
                    heightCm = heightVariation;

                    // 测量时间在事件时间前0-60分钟
                    long measureTimeMillis = eventTime - TimeUnit.MINUTES.toMillis(random.nextInt(60));
                    measureTime = sdf.format(new Date(measureTimeMillis));

                    // 90%的数据标记为有效
                    isValid = random.nextDouble() < VALID_DATA_RATIO;
                }

                // 构建JSON记录
                String userBehaviorJson = String.format(
                        "{\"user_id\":\"%s\",\"behavior_type\":\"%s\",\"item_id\":\"%s\"," +
                                "\"category_id\":\"%s\",\"time_stamp\":\"%s\",\"source\":\"%s\"," +
                                "\"birthday\":\"%s\",\"gender_input\":\"%s\",\"height_input\":%.1f," +
                                "\"weight_input\":%.1f,\"update_time\":\"%s\",\"profile_source\":\"%s\"," +
                                "\"device_type\":\"%s\",\"weight_kg\":%s,\"height_cm\":%s," +
                                "\"measure_time\":%s,\"is_valid\":%b}",
                        userId, behavior, itemId, categoryId, sdf.format(new Date(eventTime)), source,
                        profile.birthday, profile.gender, profile.height, profile.weight,
                        sdf.format(new Date(profile.updateTime)), profile.profileSource,
                        deviceType,
                        (weightKg == null ? "null" : String.format("%.1f", weightKg)),
                        (heightCm == null ? "null" : String.format("%.1f", heightCm)),
                        (measureTime == null ? "null" : "\"" + measureTime + "\""),
                        isValid
                );

                // 发送数据
                ctx.collect(userBehaviorJson);

                // 进度显示
                if ((i + 1) % 100 == 0) {
                    System.out.println("已生成 " + (i + 1) + " 条记录");
                }

                // 添加随机延迟
                Thread.sleep(random.nextInt(100));
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}